package com.bw.gmall.realtime.dwd.db.app;

import com.bw.gmall.realtime.common.base.BaseSqlApp;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;


public class DwdStorePopularity extends BaseSqlApp {
    public static void main(String[] args) {
        new DwdStorePopularity().start(Constant.TOPIC_DB, Constant.TOPIC_DWD_STORE_POPULARITY, 4, 10013);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv, String groupId) {
        //1、读取ODS数据
        readOdsDb(tableEnv, Constant.TOPIC_DWD_STORE_POPULARITY);
        //2、筛选出加购和支付相关的数据
        Table storePopularityData = getStorePopularityData(tableEnv);
        //3.创建kafka表
        createKafkaTable(tableEnv);
        //4、写出Kafka
        storePopularityData.insertInto(Constant.TOPIC_DWD_STORE_POPULARITY).execute();
    }

    private static void createKafkaTable(StreamTableEnvironment tableEnv) {
        tableEnv.executeSql("create table " + Constant.TOPIC_DWD_STORE_POPULARITY + "(\n" +
                "user_id string,\n" +
                "is_cart_add string,\n" +
                "is_paid string,\n" +
                "total_amount decimal(16, 2),\n" +
                "ts bigint\n" +
                ")" + SQLUtil.getKafkaSinkSQL(Constant.TOPIC_DWD_STORE_POPULARITY));
    }

    private static Table getStorePopularityData(StreamTableEnvironment tableEnv) {
        return tableEnv.sqlQuery("select\n" +
                "`data` ['user_id'] user_id, \n" +
                "case when `table` = 'cart_info' and (`type`='insert' or (`type`='update' and `old`['sku_num'] is not null and cast(`data`['sku_num'] as bigint) > cast(`old`['sku_num'] as bigint) )) then '1' else '0' end as is_cart_add, \n" +
                "case when `table` = 'payment_info' and `type`='insert' then '1' else '0' end as is_paid, \n" +
                "case when `table` = 'payment_info' and `type`='insert' then cast(`data`['total_amount'] as decimal(16, 2)) else 0 end as payment_amount, \n" +
                "ts \n" +
                "from topic_db\n" +
                "where `database`='gmall' and (`table`='cart_info' or `table`='payment_info')");
    }
}